package com.young.filter;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.MessageSelector;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;

import java.nio.charset.StandardCharsets;
import java.util.List;

/**
 * 生产者
 * 过滤消息有两种方式
 * 1、消费者生产消息的时候根据不同的需求将消息发送到不同的Tag, 消费者在订阅消息的时候通过Tag进行消息过滤
 * 2、消费者通过MessageSelector.bySql来进行过滤
 *
 * 如果出现 CODE: 1  DESC: The broker does not support consumer to filter message by SQL92 错误
 * 则需要修改 broker对应的properties配置文件 加上 "enablePropertyFilter=true" 开启sql 查询的功能
 */
public class FilterConsumer {
    public static void main(String[] args) throws MQClientException {
        // 1、创建消费者实例，并传递消费者组名称（推送模型的消费者组）
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("filter_consumer_group");
        // 2、消费者设组设置nameServer地址
        consumer.setNamesrvAddr("192.168.188.101:9876;192.168.188.102:9876;");
        // 3、订阅topic
        consumer.subscribe("filter_topic", MessageSelector.bySql("a between 0 and 3"));
        // 4、注册消息监听者，监听并处理消息
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                list.forEach(message -> {
                    String messageBody = new String(message.getBody(),StandardCharsets.UTF_8);
                    System.out.println("接收到消息, topic:"+message.getTopic()+"Tags："+message.getTags()+"消息内容："+messageBody);
                });
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        // 5、启动消费者
        consumer.start();

        System.out.println("消费者已启动.......");

    }
}
